package qianting.news.behavior.controller;

import org.apache.commons.lang3.StringUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import qianting.news.cache.CacheService;
import qianting.news.common.entity.State;

@RestController
@RequestMapping("/")
public class TestController {

    @Autowired
    private KafkaTemplate<String,String> kafkaTemplate;

    @GetMapping("producer")
    public String producer(String msg){

        kafkaTemplate.send("test_in",msg);
        return "ok";

    }

    @KafkaListener(topics = {"test_out"})
    public void onMessage(String msg){
        if(!StringUtils.isEmpty(msg)){
            System.out.println(msg);
        }
    }

    @KafkaListener(topics = {"test_out"})
    public void listen(ConsumerRecord<?, ?> record){
        System.out.println(record.key()+":"+record.value());
    }

    @KafkaListener(topics = {"logs"})
    public void listen1(ConsumerRecord<?, ?> record){
        System.out.println(record.key()+":"+record.value());
    }


    //测试redis缓存
    @Autowired
    private CacheService cacheService;
    @GetMapping("testRedis")
    public State testRedis(String msg){
        cacheService.set("msg",msg);
        String s = cacheService.get("msg");
        return State.succ(s);
    }

}
